package iot;

import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.Session;
import com.datastax.driver.mapping.MapperFactory;
import hbase.template.HBaseAsyncTemplate;
import hbase.template.HBaseConnectionFactory;
import hbase.template.HBaseTemplate;
import iot.queue.DataQueueDispatcher;
import netty.session.SessionManager;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Lazy;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.Executors;

/**
 * Created by dave on 18-7-21 下午5:11.
 */
@Configuration
public class ServerConfig {
    @Value("${hbase.quorum}")
    private String quorum;
    @Value("${hbase.port}")
    private int port;
    @Value("${cassandra.host:127.0.0.1}")
    private String cassandraHost;

    @Bean
    public TaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(Runtime.getRuntime().availableProcessors());
        executor.setMaxPoolSize(Runtime.getRuntime().availableProcessors() * 2);
        executor.setQueueCapacity(512);
        return executor;
    }

    @Bean
    @Lazy
    public HBaseConnectionFactory connectionFactory() {
        org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", quorum);
        conf.set("hbase.zookeeper.property.clientPort", String.valueOf(port));
        return new HBaseConnectionFactory(conf);
    }

    @Bean
    @Lazy
    @ConditionalOnExpression("${hbase.async:false}")
    public HBaseAsyncTemplate hBaseAsyncTemplate(HBaseConnectionFactory connectionFactory) {
        return new HBaseAsyncTemplate(connectionFactory, Executors.newCachedThreadPool());
    }

    @Bean
    @Lazy
    @ConditionalOnExpression("!${hbase.async:false}")
    public HBaseTemplate hBaseTemplate(HBaseConnectionFactory connectionFactory) {
        return new HBaseTemplate(connectionFactory, Executors.newCachedThreadPool());
    }

    @Bean
    @Lazy
    public Session cassandraSession() {
        return Cluster.builder().addContactPoint(cassandraHost)
                .build().connect();
    }

    @Bean
    @Lazy
    public MapperFactory mapperFactory(Session session) {
        return new MapperFactory(session);
    }

    @Bean
    public SessionManager sessionManager() {
        return new SessionManager();
    }

    @Bean
    public MessageProcessor messageProcessor(SessionManager sessionManager, DataDispatcher dispatcher) {
        return new MessageProcessor(sessionManager, dispatcher);
    }
}
